package com.zlsf.publish.local;

import cn.hutool.core.util.HexUtil;
import com.zlsf.flv.FlvEncoder;
import io.vertx.core.buffer.Buffer;
import io.vertx.ext.web.RoutingContext;
import lombok.extern.slf4j.Slf4j;

import java.util.LinkedList;
import java.util.concurrent.atomic.AtomicLong;

@Slf4j
public abstract class SubscriberLocal extends Thread {

  static final AtomicLong SEQUENCE = new AtomicLong(0L);
  private long id;
  private String tag;
  private Object lock;
  private RoutingContext context;
  protected LinkedList<byte[]> messages;

  public SubscriberLocal(String tag, RoutingContext ctx) {
    this.tag = tag;
    this.context = ctx;
    this.lock = new Object();
    this.messages = new LinkedList<byte[]>();

    this.id = SEQUENCE.getAndAdd(1L);
  }

  public long getId() {
    return this.id;
  }

  public String getTag() {
    return this.tag;
  }

  public abstract void onVideoData(long timeoffset, byte[] data, FlvEncoder flvEncoder);

  public abstract void onAudioData(long timeoffset, byte[] data, FlvEncoder flvEncoder);

  public void enqueue(byte[] data) {
    if (data == null) return;
    synchronized (lock) {
      messages.addLast(data);
      lock.notify();
    }
  }

  public void run() {
    while (!this.isInterrupted()) {
      try {
        byte[] data = take();
        if (data != null) send(data);
      } catch (Exception ex) {
        //销毁线程时，如果有锁wait就不会销毁线程，抛出InterruptedException异常
        log.error("send failed", ex);
        break;
      }
    }
    log.info("subscriber closed");
  }

  protected byte[] take() {
    byte[] data = null;
    try {
      synchronized (lock) {
        while (messages.isEmpty()) {
          lock.wait(100);
          if (this.isInterrupted()) return null;
        }
        data = messages.removeFirst();
      }
      return data;
    } catch (Exception ex) {
      this.interrupt();
      return null;
    }
  }

  public void close() {
    this.interrupt();
  }

  private static int count = 0;

  public void send(byte[] message) {
    count++;
    log.info("count:{} ", count);
    log.info(HexUtil.encodeHexStr(message));
    context.response().write(Buffer.buffer(message));
  }
}
